package com.shujia.MapReduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Demo01WordCount {
    // Map任务

    /**
     * 继承Mapper类，指定输入的K-V的类型以及输出的K—V的类型
     * 然后重写map方法
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        /**
         * key 偏移量
         * value 一行数据
         * context MR任务运行是的上下文环境，可以获取当前运行的配置等其他信息
         *          主要用于将map端构建好的K-V进行输出
         */
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            // 自己的map逻辑
            // 按分隔符切分出每一个单词
            String[] splits = value.toString().split(",");
            for (String word : splits) {
                // 以单词作为key，1作为value，构建好输出的k-v
                // 再通过context输出
                /**
                 * Java,C,C++,Shell,SQL,Scala,Python
                 * 输出：Java,1
                 *      C,1
                 *      C++,1
                 *      Shell,1
                 *      SQL,1
                 *      Scala,1
                 *      Python,1
                 */
                Text outputKey = new Text(word);
                IntWritable outputValue = new IntWritable(1);
                context.write(outputKey, outputValue);
            }

        }
    }

    // Reduce任务

    /**
     * 继承Reducer类，指定输入的K-V的类型(同Map端的输出的K-V类型保持一致)以及输出的K—V的类型
     * 然后重写reduce方法
     */
    public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        /**
         * key 经过shuffle过程，相同的Key会进入同一个reduce
         * values 相同的Key对应的Value的一个"集合"
         * context 主要用于输出数据到HDFS
         */
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int sum = 0; // 用于保存最后的结果
            for (IntWritable value : values) {
                sum += value.get();
            }
            // 将最后的结果输出
            context.write(key, new IntWritable(sum));
        }
    }

    // Driver程序，主要负责配置及提交任务
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://master:9000");
        // 创建一个MapReduce的Job
        Job job = Job.getInstance(conf);
        // 配置任务
        // 设置任务运行的名字
        job.setJobName("Demo01WordCount");
        // 设置任务运行哪一个类
        job.setJarByClass(Demo01WordCount.class);
        // 设置Reduce的数量，默认1，最终生成文件的数量同Reduce的数量一致
        job.setNumReduceTasks(3);

        // 配置Map端
        // 指定Map任务运行哪一个类
        job.setMapperClass(MyMapper.class);
        // 配置Map端输出的Key的类型
        job.setMapOutputKeyClass(Text.class);
        // 配置Map端输出的Value的类型
        job.setMapOutputValueClass(IntWritable.class);

        // 配置Reduce端
        // 指定Reduce任务运行哪一个类
        job.setReducerClass(MyReducer.class);
        // 配置Reduce端输出的Key的类型
        job.setOutputKeyClass(Text.class);
        // 配置Reduce端输出的Value的类型
        job.setOutputValueClass(IntWritable.class);

        // 配置输入输出路径

        FileInputFormat.addInputPath(job, new Path("/data/wc/input"));
//        FileInputFormat.addInputPath(job, new Path("/data/wc/input1"));
//        FileInputFormat.addInputPath(job, new Path("/data/wc/input2"));

        Path path = new Path("/data/wc/output");
        FileSystem fs = FileSystem.get(conf);
        // 判断输出路径是否存在，存在即删除
        if (fs.exists(path)) {
            fs.delete(path, true);
        }
        // 输出路径已存在会报错
        FileOutputFormat.setOutputPath(job, path);


        // 等待任务运行完成
        job.waitForCompletion(true);

    }

    /**
     * 1、将words.txt上传至虚拟机并使用HDFS命令上传至HDFS
     *  hdfs dfs -mkdir /data/wc/input
     *  hdfs dfs -put words.txt /data/wc/input
     * 2、将代码通过maven的package打包成jar包，并上传至虚拟机
     * 3、使用命令提交任务
     *  hadoop jar Hadoop-1.0.jar com.shujia.MapReduce.Demo01WordCount
     *
     *  查看日志： 在任务运行时会自动生成一个applicationId
     *  yarn logs -applicationId application_1647850049046_0004
     *  也可以通过historyserver去查看，因为任务真正是运行在NodeManager中，日志可能会分散
     *  historyserver可以负责从NodeManager中收集日志到Master中方便查看日志
     *  启动historyServer：在Master上启动即可
     *      mr-jobhistory-daemon.sh start historyserver
     *  http://master:19888
     */
}
